package com.leilei.broadcaststate;

import com.alibaba.fastjson.JSON;
import com.leilei.broadcaststate.common.Location;
import com.leilei.broadcaststate.common.RabbitConfigUtil;
import com.leilei.broadcaststate.common.RabbitMqSource;
import com.leilei.broadcaststate.common.VehicleOrgRelationSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/24 21:36
 * @desc 广播配置或规则路由到下游所有task
 * 下游的 task 接收这些配置、规则并保存为 BroadcastState, 将这些配置应用到另一个数据流的计算中 。
 * ex: 动态更新计算规则 、实时增加额外字段
 */
public class FlinkBroadcaststate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        DataStreamSource<String> mqLocationSource = env.addSource(new RabbitMqSource(RabbitConfigUtil.getConnectionConfig(),
                "paas-common-calc-stream.location_alarm",
                "paas-trans-enterprise-platform-data.fanout.location_test"));
        SingleOutputStreamOperator<Location> locationSource = mqLocationSource.map(e -> JSON.parseObject(e, Location.class));
        final DataStreamSource<Tuple2<Integer, List<Integer>>> relationSource = env.addSource(new VehicleOrgRelationSource());

        MapStateDescriptor<Integer, List<Integer>> descriptor = new MapStateDescriptor<>("relationState",
                Types.INT, Types.LIST(Types.INT));
        //-2.将配置流根据状态描述器广播出去,变成广播状态流
        final BroadcastStream<Tuple2<Integer, List<Integer>>> broadcastStream = relationSource.broadcast(descriptor);

        //-3.将事件流和广播流进行连接
        final BroadcastConnectedStream<Location, Tuple2<Integer, List<Integer>>> connect = locationSource.connect(broadcastStream);
        //-4.处理连接后的流-根据配置流补全事件流中的信息
        final DataStream<Location> process = connect.process(new LocationProcessFunction(descriptor));
        process.print();

        env.execute();
    }
}
